package com.gukun.springboot.tdengine.service.backuprecover.backup;

import cn.hutool.core.date.BetweenFormatter;
import cn.hutool.core.date.DateUtil;
import com.gukun.springboot.tdengine.domain.backuprecover.BackupRecoverLogEntity;
import com.gukun.springboot.tdengine.domain.backuprecover.BackupRecoverTaskEntity;
import com.gukun.springboot.tdengine.domain.exception.ServerException;
import com.gukun.springboot.tdengine.domain.exception.ServiceException;
import com.gukun.springboot.tdengine.domain.exception.util.ServiceExceptionUtil;
import com.gukun.springboot.tdengine.pojo.DBConfig;
import com.gukun.springboot.tdengine.service.backuprecover.BackupRecoverLogService;
import com.gukun.springboot.tdengine.service.backuprecover.BackupRecoverTaskService;
import com.gukun.springboot.tdengine.service.backuprecover.config.TDEngineConnPool;
import com.gukun.springboot.tdengine.service.backuprecover.config.ThreadPool;
import com.gukun.springboot.tdengine.service.backuprecover.entity.LogRecorder;
import com.gukun.springboot.tdengine.service.backuprecover.entity.Table;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.RateLimiter;
import org.apache.ibatis.javassist.SerialVersionUID;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhang
 */
@Component
public class BackupExecutor {

    @Autowired
    @Lazy
    private BackupRecoverLogService backupRecoverLogService;
    @Autowired
    @Lazy
    private BackupRecoverTaskService backupRecoverTaskService;

    //获取数据库表格信息
    private static final String GET_TABLE_NAME = "select table_name,stable_name from INFORMATION_SCHEMA.ins_tables where db_name = '%s'";
    //无界队列，存储表格信息
    private final BlockingQueue<Table> tableNameQueue = new LinkedBlockingQueue<>();

    private static final ConcurrentHashMap<String,CompletableFuture<?>> FUTURE_MAP = new ConcurrentHashMap<>();

    public void execute(BackupRecoverTaskEntity task) throws Exception {
        if(!FUTURE_MAP.isEmpty()){
            System.out.println("当前任务数：" + FUTURE_MAP.size());
            throw ServiceExceptionUtil.exception0(200,"当前有任务执行中，请稍后再试");
        }
        int operateTarget = task.getOperateTarget();
        Thread thread = switch (operateTarget) {
            case 1 -> new Thread(() -> {
                //数据库数据备份
                try {
                    executeDbData(task);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            case 2 -> new Thread(() -> {
                //数据库表格备份
                try {
                    executeDbTable(task);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            default -> throw new RuntimeException("备份目标错误");
        };
        thread.start();
    }
    /**
     * 目前已经备份的表格数量
     */
    private AtomicLong tableNum;
    /**
     * 表格总数
     */
    private AtomicLong tableCount;

    /**
     * 获取表格备份进度
     */
    public float getTableBackupProgress(){
        if(tableNum!=null && tableCount!=null){
            if(tableCount.get() == 0){
                return 0;
            }
            return Float.parseFloat(String.format("%.2f",tableNum.get()*100.0/tableCount.get()));
        }
        return 0;
    }

    public float getTableDataBackupProgress(){
        float progress = 0;
        if(tableCount !=null && tableCount.get() != 0){
            progress = (tableCount.get() - tableNameQueue.size()) * 100/(float)tableCount.get();
        }
        return Float.parseFloat(String.format("%.2f",progress));
    }

    public void executeDbData(BackupRecoverTaskEntity task) throws InterruptedException {
        Date startTime = new Date();
        //任务状态修改
        backupRecoverTaskService.updateTaskStatusStart(task);
        String startStr = DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss");
        LogRecorder logRecorder = new LogRecorder(task,backupRecoverLogService,startStr);
        logRecorder.appendMessage("任务初始化");
        logRecorder.save();
        DBConfig database = task.getDbConfig();
        String jdbcUrl = database.getJdbcUrl();
        int threadNum = task.getThreadNum();
        logRecorder.appendMessage("执行任务线程数：" + threadNum);
        HikariDataSource tdEngineConnPool = null;
        try{
            tdEngineConnPool  = TDEngineConnPool.getHikariDataSource(jdbcUrl,threadNum);
        }catch (Exception e){
            logRecorder.appendMessage("获取数据库连接失败：" + e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        if(tdEngineConnPool == null){
            logRecorder.appendMessage("获取数据库连接失败：" + jdbcUrl);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        String dbName = task.getDbName();
        try(Connection connection = tdEngineConnPool.getConnection()){
            getTableNames(dbName, connection);
            logRecorder.appendMessage("获取数据库表格名称成功：" + tableNameQueue.size() + "个");
        }catch (Exception e){
            tdEngineConnPool.close();
            logRecorder.appendMessage("获取数据库表格名称失败：" + e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        tableCount = new AtomicLong(tableNameQueue.size());
        String outputDir = task.getFilePath();
        if(!outputDir.endsWith("/")){
            outputDir += "/";
        }
        outputDir = outputDir + DateUtil.format(new Date(),"yyyyMMddHHmmssSSS");
        File file = new File(outputDir);
        if (!file.exists()) {
            file.mkdirs();
        }
        String finalFolder = outputDir + "/";
        logRecorder.updateFilePath(finalFolder);
        initFolder(finalFolder,tableNameQueue.size(),dbName);
        logRecorder.appendMessage("开始备份数据库数据");
        logRecorder.save();
        //多线程处理数据
        ThreadPoolExecutor threadPoolExecutor = ThreadPool.threadPoolExecutor(threadNum,threadNum);
        CompletableFuture<?>[] futures = new CompletableFuture[threadNum];
        CompletableFuture<Void> future = null;
        AtomicLong dataNum = new AtomicLong(0);
        for (int i=0;i<threadNum;i++) {
            future = CompletableFuture.runAsync(new BackupDbDataTask(tableNameQueue,task,finalFolder,dataNum,tdEngineConnPool),threadPoolExecutor);
            futures[i]=future;
        }
        try{
            CompletableFuture<?> all = CompletableFuture.allOf(futures);
            FUTURE_MAP.put(task.getId(),all);
            all.get();
        }catch (CancellationException e){
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            threadPoolExecutor.shutdownNow();
            logRecorder.appendMessage("数据库数据备份失败:任务手动停止");
            logRecorder.saveFailed();
        } catch(Exception e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            logRecorder.appendMessage("数据库数据备份失败:"+e.getMessage());
            logRecorder.saveFailed();
        }
        threadPoolExecutor.shutdown();
        if(!threadPoolExecutor.awaitTermination(1000,TimeUnit.MILLISECONDS)){
            logRecorder.appendMessage("线程池关闭失败");
        }else{
            logRecorder.appendMessage("线程池关闭");
        }
        tdEngineConnPool.close();
        Date endTime = new Date();
        String endTimeStr = DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss");
        logRecorder.updateEndTime(endTimeStr);
        logRecorder.appendMessage("数据库数据备份完成");
        logRecorder.appendMessage("备份耗时:"+DateUtil.formatBetween(startTime, endTime, BetweenFormatter.Level.MILLISECOND));
        logRecorder.appendMessage("备份数据条数："+dataNum);
        File logFile = new File(finalFolder + "log.txt");
        logRecorder.appendMessage("开始生成日志文件");
        try{
            if(logFile.createNewFile()){
                BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(logFile));
                bufferedWriter.write(String.valueOf(dataNum.get()));
                bufferedWriter.flush();
                bufferedWriter.close();
                logRecorder.appendMessage("生成日志文件成功");
            }else{
                logRecorder.appendMessage("生成日志文件失败");
            }
        }catch (Exception e){
            e.printStackTrace();
            logRecorder.appendMessage("生成日志文件失败");
        }
        if(logRecorder.isRunning()){
            logRecorder.saveSuccess();
        }else{
            logRecorder.saveFailed();
        }
        backupRecoverTaskService.updateTaskStatusStop(task);
        FUTURE_MAP.clear();
        tableCount = null;
        tableNum = null;
        tableNameQueue.clear();
    }

    public void stopTask(String id){
        CompletableFuture<?> futures = FUTURE_MAP.get(id);
        if(futures != null){
            futures.cancel(true);
        }else{
            throw ServiceExceptionUtil.exception0(500, "任务初始化未完成");
        }
    }

    public void executeDbTable(BackupRecoverTaskEntity task) throws InterruptedException {
        Date startTime = new Date();
        String startTimeStr = DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss");
        LogRecorder logRecorder = new LogRecorder(task,backupRecoverLogService,startTimeStr);
        logRecorder.appendMessage("任务初始化");
        logRecorder.save();
        backupRecoverTaskService.updateTaskStatusStart(task);
        String filePath = task.getFilePath();
        if(!filePath.endsWith("/")){
            filePath = filePath + "/";
        }
        filePath += DateUtil.format(new Date(),"yyyyMMddHHmmssSSS");
        File file = new File(filePath);
        if(!file.exists()){
            file.mkdirs();
        }
        filePath = filePath + "/";
        logRecorder.updateFilePath(filePath);
        String dbName = task.getDbName();
        DBConfig database = task.getDbConfig();
        String[] stables = new String[]{"tfloat","tbool","tint","tdouble"};
        HikariDataSource tdEngineConnPool = null;
        try{
            tdEngineConnPool = TDEngineConnPool.getHikariDataSource(database.getJdbcUrl(),4);
        }catch (Exception e){
            logRecorder.appendMessage("获取数据库连接失败：" + e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        if(tdEngineConnPool == null){
            logRecorder.appendMessage("获取数据库连接失败：" + database.getJdbcUrl());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        logRecorder.appendMessage("开始备份数据库表格："+dbName);
        logRecorder.appendMessage("开始获取表格数量");
        try(Connection connection = tdEngineConnPool.getConnection()){
            getTableCount(dbName,connection);
        }catch (Exception e){
            logRecorder.appendMessage("获取表格数量失败：" + e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            return;
        }
        logRecorder.appendMessage("获取表格数量成功");
        logRecorder.save();
        File logFile = new File(filePath + "log.txt");
        try{
            if(logFile.createNewFile()){
                BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(logFile));
                bufferedWriter.write(String.valueOf(tableCount.get()));
                bufferedWriter.flush();
                bufferedWriter.close();
            }else{
                logRecorder.appendMessage("创建日志文件失败");
                logRecorder.saveFailed();
                backupRecoverTaskService.updateTaskStatusStop(task);
                tdEngineConnPool.close();
                return;
            }
        }catch (Exception e){
            logRecorder.appendMessage("创建日志文件失败：" + e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            return;
        }
        ThreadPoolExecutor threadPoolExecutor = ThreadPool.threadPoolExecutor(4,4);
        CompletableFuture<?>[] futures = new CompletableFuture[4];
        CompletableFuture<Void> future = null;
        tableNum = new AtomicLong(0);
        for(int i=0;i<4;i++){
            future = CompletableFuture.runAsync(new BackupDbTableTask(dbName,stables[i],tdEngineConnPool,tableNum,filePath),threadPoolExecutor);
            futures[i] = future;
        }
        try{
            CompletableFuture<Void> all = CompletableFuture.allOf(futures);
            FUTURE_MAP.put(task.getId(),all);
            all.get();
        }catch (CancellationException e){
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            threadPoolExecutor.shutdownNow();
            logRecorder.appendMessage("数据库表格备份失败:任务手动停止");
            logRecorder.saveFailed();
        } catch (Exception e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            threadPoolExecutor.shutdownNow();
            logRecorder.appendMessage("数据库表格备份失败:"+e.getMessage());
            logRecorder.saveFailed();
        }
        threadPoolExecutor.shutdown();
        if(!threadPoolExecutor.awaitTermination(1000,TimeUnit.MILLISECONDS)){
            logRecorder.appendMessage("线程池关闭失败");
        }else{
            logRecorder.appendMessage("线程池关闭");
        }
        tdEngineConnPool.close();
        Date endTime = new Date();
        String endTimeStr = DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss");
        logRecorder.updateEndTime(endTimeStr);
        logRecorder.appendMessage("数据库表格备份完成");
        logRecorder.appendMessage("备份耗时:"+DateUtil.formatBetween(startTime, endTime, BetweenFormatter.Level.MILLISECOND));
        logRecorder.appendMessage("备份数据表格数："+tableNum.get());
        if(logRecorder.isRunning()){
            logRecorder.saveSuccess();
        }else{
            logRecorder.saveFailed();
        }
        tableCount = null;
        tableNum = null;
        FUTURE_MAP.clear();
        backupRecoverTaskService.updateTaskStatusStop(task);
    }

    //25000个文件一个文件夹
    private void initFolder(String outputDir, int size,String dbName) {
        int num = size / 20000;
        if(num % 20000 != 0){
            num++;
        }
        for (int i = 0; i < num; i++) {
            String path = outputDir + dbName+"_" + i;
            File file = new File(path);
            if (!file.exists()) {
                file.mkdir();
            }
        }
    }

    public void getTableCount(String dbName, Connection connection){
        String sql = "select count(*) as num from INFORMATION_SCHEMA.ins_tables where db_name = '%s'";
        try (Statement statement = connection.createStatement()) {
            ResultSet resultSet = statement.executeQuery(String.format(sql, dbName));
            while  (resultSet.next()) {
                tableCount = new AtomicLong(resultSet.getLong("num"));
            }
            resultSet.close();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    public void getTableNames(String dbName, Connection connection) {
        String tableName = null;
        String stableName = null;
        int index = 0;
        //限速
        RateLimiter rateLimiter = RateLimiter.create(50000);
        try (Statement statement = connection.createStatement()) {
            statement.setFetchSize(2500);
            ResultSet resultSet = statement.executeQuery(String.format(GET_TABLE_NAME, dbName));
            while  (resultSet.next()) {
                rateLimiter.acquire();
                tableName = resultSet.getString("table_name");
                stableName = resultSet.getString("stable_name");
                tableNameQueue.put(new Table(tableName, stableName,index));
                index++;
            }
            resultSet.close();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("获取数据库表格名称失败：" + e.getMessage());
        }
    }
}
